package com.atguigu.gmall.gateway.filter;

import lombok.extern.slf4j.Slf4j;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Scheduler;
import reactor.core.scheduler.Schedulers;

import java.time.Duration;

/**
 * @author lfy
 * @Description
 * @create 2023-07-28 14:49
 */
@Slf4j
public class ReactiveMainTest {

    /**
     * 响应式编程认为
     * 1、万物皆流；
     * 2、万物皆可异步；
     * 3、流有两种：
     *         0|1:  Mono<User>
     *      很多数据： Flux<User>
     * 4、数据皆用发布订阅模式
     * @param args
     */
    public static void aaaamain(String[] args) throws InterruptedException {

        //1、构造一个流; 构造一个数据发布者
        Mono<String> just = Mono.just("1");
        log.info("线程："+Thread.currentThread().getName());

        //2、消费流；构造一个消费者
        just.subscribe(item->{
            log.info("张三1拿到数据："+item+"="+Thread.currentThread().getName());
        });


        just.subscribe(item->{
            log.info("张三2拿到数据："+item+"="+Thread.currentThread().getName());
        });


        just.subscribe(item->{
            log.info("张三3拿到数据："+item+"="+Thread.currentThread().getName());
        });


        //===========Flux=========
        Flux<Long> flux = Flux.interval(Duration.ofSeconds(1));


        //背压：基于消费者的消费能力，慢慢处理，而不要压垮业务。
        flux.subscribe(item->{
            log.info("A："+item);
        });

        flux.subscribe(item->{
            try {
                Thread.sleep(3000);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            log.info("B："+item);
        });


        Thread.sleep(10000000);
    }



    public String hello() throws InterruptedException {

        Thread.sleep(3000);
        return "hello world";
    }


    /**
     * 响应式 = 异步（Callable） + 消息发布订阅（Mono、Flux）
     * @return
     */
    public Mono<String> helloMono(){
        return Mono.fromCallable(()->hello());
    }

    public Mono<Void> worldMono(Mono<String> mono){
        mono.subscribeOn(Schedulers.parallel())
                .subscribe(item->{
                    try {
                        Thread.sleep(3000);
                    } catch (InterruptedException e) {
                        throw new RuntimeException(e);
                    }
                    item += "!!!!abc";
                    log.info("新数据："+item);
                });
        return Mono.empty();
    }

    public static void main(String[] args) throws InterruptedException {
        ReactiveMainTest test = new ReactiveMainTest();

        Mono<String> mono = test.helloMono();
        test.worldMono(mono);
        log.info("abcccc");

//        //1、Mono是一个发布者的数据流
//        Mono<String> mono = test.helloMono();  //非阻塞调用
//        log.info("abc");
//
//
//        //当前线程订阅
//        mono.subscribeOn(Schedulers.parallel()) //使用并发订阅。开一个新线程消费数据
//            .subscribe(item->{  //默认是当前线程订阅
//                log.info("数据："+item);
//            });
//
//        log.info("def");

        Thread.sleep(10000000);
    }







}
